#include <algorithm>
#include <queue>
#include <map>
#include <initializer_list>
#include "ObjectController.h"
#include "PublicFunction.h"

#include "StoreObjectMerge.h"
#include "Config.h"

using namespace std;
namespace gitstore {
CObjectController::CObjectController() {
    auto config = CGlobalConfig::getInstance()->getConfigContent();
    _io_worker = make_shared<CEtcdMetaDataIO>(config["etcd_url"].asString());
    switch(static_cast<StoreSystemType>(config["store_type"].asInt())) {
    case StoreSystemType::Ceph:
        _obj_io_worker = make_unique<CCephObjectIO>(config["ceph_cluster_name"].asString(), config["ceph_user_name"].asString(), config["ceph_conf_path"].asString(), config["ceph_object_pool_name"].asString());
        break;
    default:
        _obj_io_worker = make_unique<CLinuxObjectIOBase>();
        break;
    }

    _ds_op = make_unique<CObjectOperation>();
    _br_op = make_unique<CBranchOperation>();
    _node_op = make_unique<CNodeOperation>();
}

CObjectController::CObjectController(std::shared_ptr<CMetaDataIOBase> io_worker)
    : _io_worker(io_worker){
    auto config = CGlobalConfig::getInstance()->getConfigContent();
    switch(static_cast<StoreSystemType>(config["store_type"].asInt())) {
    case StoreSystemType::Ceph:
        _obj_io_worker = make_unique<CCephObjectIO>(config["ceph_cluster_name"].asString(), config["ceph_user_name"].asString(), config["ceph_conf_path"].asString(), config["ceph_object_pool_name"].asString());
        break;
    default:
        _obj_io_worker = make_unique<CLinuxObjectIOBase>();
        break;
    }

    _ds_op = make_unique<CObjectOperation>();
    _br_op = make_unique<CBranchOperation>();
    _node_op = make_unique<CNodeOperation>();
}

CObjectController::~CObjectController() {

}

//ErrorStatus CObjectController::beginTransaction(const std::string& obj_name) {
////    return _io_worker->begin(obj_name);
//}

//ErrorStatus CObjectController::commitTransaction(const std::string& obj_name) {
////    return _io_worker->commit(obj_name);
//}

//ErrorStatus CObjectController::rollbackTransaction(const std::string& obj_name) {
////    return _io_worker->rollback(obj_name);
//}

ErrorStatus CObjectController::isObjectExist(const std::string& obj_name, bool &b_exist) {
    //    vector<string> vds;
    //    if ( Success != _io_worker->listObject(vds) )
    //        return DatasetListError;
    //    auto iter = std::find(vds.begin(), vds.end(), obj_name);
    //    b_exist = (iter != vds.end());
    Object obj;
    int revision = NO_REVISION;
    auto res = getObjectInfo(obj_name, obj, revision);
    b_exist = ( Success == res );
    return Success;
}

ErrorStatus CObjectController::getObjectInfo(const std::string& obj_name, Object &object, int &revision) {
    //    bool b_exist = false;
    //    auto res = this->isObjectExist(obj_name, b_exist);
    //    if ( Success != res )
    //        return res;
    //    if ( !b_exist )
    //        return DatasetNotExist;

    //    return _ds_op->load(this->_io_worker, obj_name, object, revision);
    auto res = _io_worker->loadObject(obj_name, object, revision);
    if ( Success != res ) {
        return DatasetLoadError;
    }
    return res;
}

ErrorStatus CObjectController::listObject(std::vector<std::string> &vds) {
    return _io_worker->listObject(vds);
}

ErrorStatus CObjectController::createObject(const std::string& obj_name, const std::string& ds_note,
                                            StoreObjectType type, StoreObjectType subtype,
                                            const std::string& br_name, const std::string& br_note,
                                            const std::string& commit_note, const StoreObject& value,
                                            int& revision_obj, int& revision_branch) {
    bool b_exist = false;
    auto res = isObjectExist(obj_name, b_exist);
    if ( b_exist )
        return DatasetAlreadyExist;
    string version = sha256(value.getSerializedValue());
    Node node;
    res = _node_op->create(obj_name, version, commit_note, static_cast<int>(type), vector<string>{}, node);
    if ( Success != res )
        return res;
    Object object;
    if ( type == StoreObjectType::String || type == StoreObjectType::Blob ) {
        subtype = type;
    }
    if ( subtype == StoreObjectType::List || subtype == StoreObjectType::Map ) {
        return SubTypeOnlyCanBeStringOrBlob;
    }
    res = _ds_op->create(obj_name, ds_note, br_name, br_note, static_cast<int>(type), static_cast<int>(subtype), version, object);
    if ( Success != res )
        return res;
    Branch branch;
    res = _br_op->create(obj_name, br_name, br_note, version, static_cast<int>(type), branch);
    if ( Success != res )
        return res;

    int revision_node = NO_REVISION;
    res = _io_worker->createNode(node, revision_node);
    if ( Success != res ){
        return res;
    }
    res = _obj_io_worker->setValue(version, value);
    if ( Success != res ){
        return res;
    }
    res = _io_worker->createBranch(branch, revision_branch);
    if ( Success != res ){
        return res;
    }
    res = _io_worker->createObject(object, revision_obj);
    if ( Success != res ){
        return res;
    }
    return Success;
}

ErrorStatus CObjectController::getBranchInfo(const Object &object, const std::string& br_name,
                                             Branch &branch, int& revision) {
    //    bool b_exist = false;
    //    auto res = this->isBranchExist(object.name, br_name, b_exist);
    //    if ( Success != res )
    //        return res;
    //    if ( !b_exist )
    //        return BranchNotExist;
    auto res = _io_worker->loadBranch(object.name, br_name, branch, revision);
    //    auto res = _br_op->load(this->_io_worker, object.name, br_name, branch, revision);
    if ( Success != res ) {
        return BranchLoadError;
    }
    return res;
}

ErrorStatus CObjectController::getBranchInfo(const std::string& obj_name, const std::string& br_name,
                                             Branch &branch, int& revision) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res )
        return res;
    res = getBranchInfo(object, br_name, branch, revision);
    return res;
}

ErrorStatus CObjectController::isBranchExist(const Object &object, const std::string &br_name, bool &b_exist) {
    //    auto iter = std::find(object.branches.begin(), object.branches.end(), br_name);
    //    b_exist = (iter != object.branches.end());
    Branch branch;
    int revision = NO_REVISION;
    auto res = getBranchInfo(object, br_name, branch, revision);
    //    auto res = _io_worker->loadBranch(object.name, br_name, branch, revision);
    b_exist = ( Success == res );
    return Success;
}

ErrorStatus CObjectController::isBranchExist(const std::string& obj_name, const std::string& br_name, bool &b_exist) {
    Object object;
    int revision_object;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res )
        return res;
    return isBranchExist(object, br_name, b_exist);
}

ErrorStatus CObjectController::getValueByVersion(const Object& object, const std::string& version,
                                                 StoreObject& value) {
    Node node;
    int revision_node = NO_REVISION;
    auto res = getNodeInfo(object, version, node, revision_node);
    if ( Success != res )
        return res;
//    value.setType(static_cast<StoreObjectType>(object.type));
    return _obj_io_worker->getValue(version, value);
}

ErrorStatus CObjectController::getValueByBranch(const std::string& obj_name, const std::string& br_name,
                                                StoreObject& value, int& revision_branch) {
    Object object;
    int revision = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision);
    if ( Success != res )
        return res;
    return getValueByBranch(object, br_name, value, revision_branch);
}

ErrorStatus CObjectController::getValueByBranch(const Object& object, const std::string& br_name,
                                               StoreObject& value, int& revision_branch) {
   Branch branch;
   auto res = getBranchInfo(object, br_name, branch, revision_branch);
   if ( Success != res )
       return res;
   return _obj_io_worker->getValue(branch.cur_version, value);
}

ErrorStatus CObjectController::getValueByVersion(const std::string& obj_name, const std::string& version,
                                                 StoreObject& value) {
    Object object;
    int revision = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision);
    if ( Success != res )
        return res;
    return getValueByVersion(object, version, value);
}


ErrorStatus CObjectController::branchObject(const std::string& obj_name, const std::string& br_refer_name,
                                            const std::string& br_new_name, const std::string& br_note,
                                            int revision_refer_branch_pre, int& revision_new_branch) {
    Object object;
    int rivison_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, rivison_object);
    if ( Success != res )
        return res;


    Branch refer_branch, new_branch;
    int revision_refer_branch = NO_REVISION;
    res = getBranchInfo(object, br_refer_name ,refer_branch, revision_refer_branch);
    //        res = _br_op->load(this->_io_worker, obj_name, br_refer_name, refer_branch);
    if ( Success != res )
        return res;
    if ( revision_refer_branch_pre != revision_refer_branch ) {
        return RevisionConflict;
    }
    //    int revision_new_branch = NO_REVISION;
    res = getBranchInfo(object, br_new_name ,refer_branch, revision_new_branch);
    if ( Success == res )
        return BranchAlreadyExist;
    res = _br_op->create(obj_name, br_new_name, br_note, refer_branch.cur_version, object.type, new_branch);
    if ( Success != res )
        return res;
    //    res = _ds_op->branchObject(object, br_refer_name, br_new_name, br_note);
    //    if ( Success != res )
    //        return res;
    //    _io_worker->begin(obj_name);
    res = _io_worker->createBranch(new_branch, revision_new_branch);//_br_op->dump(this->_io_worker, new_branch, NO_REVISION, true);
    if ( Success != res ){
        //        _io_worker->rollback(obj_name);
        return res;
    }
    //    res = _ds_op->dump(this->_io_worker, object, rivison_object, false);
    //    if ( Success != res ){
    //        _io_worker->rollback(obj_name);
    //        return res;
    //    }
    return Success;
}

//template<typename T>
//ErrorStatus CObjectController::getValueByBranch(const Object& object, const std::string& br_name,
//                                               CNodeValue<T>& value, int& revision_branch) {
//   Branch branch;
//   auto res = getBranchInfo(object, br_name, branch, revision_branch);
//   if ( Success != res )
//       return res;
//   value.setType(static_cast<StoreObjectType>(object.type));
//   return _obj_io_worker->getValue(branch.cur_version, value);
//}



ErrorStatus CObjectController::getNodeInfo(const std::string& obj_name, const std::string& version,
                                           Node& node, int& revision_node) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res ) {
        return res;
    }
    return getNodeInfo(object, version, node, revision_node);
}

ErrorStatus CObjectController::getNodeInfo(const Object &object, const std::string& version,
                                           Node &node, int& revision_node) {
    //    bool b_in = false;
    //    auto res = this->isVersionInObject(object, version, b_in);
    //    if ( Success != res )
    //        return res;
    //    if ( !b_in )
    //        return VersionNotExist;
    //    int revision_node = NO_REVISION;
    auto res = _io_worker->loadNode(object.name, version, node, revision_node);
    //    auto res = _node_op->load(this->_io_worker, object.name, version, node, revision_node);
    if ( Success != res ) {
        return VersionLoadError;
    }
    return res;
}

//template <typename T>
//ErrorStatus CObjectController::getValueByVersion(const Object& object, const std::string& version,
//                                                 CNodeValue<T>& value) {
//    Node node;
//    int revision_node = NO_REVISION;
//    auto res = getNodeInfo(object, version, node, revision_node);
//    if ( Success != res )
//        return res;
//    value.setType(static_cast<StoreObjectType>(object.type));
//    return _obj_io_worker->getValue(version, value);
//}



ErrorStatus CObjectController::getBranchCurNodeInfo(const std::string& obj_name, const std::string& br_name,
                                                    Node& node, int& revision_branch) {
    Branch branch;
    //    int revision_branch = NO_REVISION;
    auto res = getBranchInfo(obj_name, br_name, branch, revision_branch);
    if ( Success != res )
        return res;
    int revision_node = NO_REVISION;
    return getNodeInfo(obj_name, branch.cur_version, node, revision_node);
}

ErrorStatus CObjectController::listObjectBranches(const std::string& obj_name, std::vector<std::string> &vbr) {
    Object object;
    int revision = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision);
    if ( Success != res ) {
        return res;
    }
    res = _io_worker->listBranches(obj_name, vbr);
    return res;
    //    vbr.clear();
    //    vbr.reserve(object.branches.size());
    //    vbr.insert(vbr.begin(), object.branches.begin(), object.branches.end());
    //    return Success;
}

ErrorStatus CObjectController::listObjectVersions(const std::string& obj_name, std::vector<std::string> &vvs) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res ) {
        return res;
    }
    res = _io_worker->listNodes(obj_name, vvs);
    return res;
    //    vvs.clear();
    //    vvs.reserve(object.nodes.size());
    //    vvs.insert(vvs.begin(), object.nodes.begin(), object.nodes.end());
    //    return Success;
}



ErrorStatus CObjectController::checkPreviousVersion(const Branch& branch, const std::string& version_previous) {
    if ( branch.cur_version.compare(version_previous) == 0 ) {
        return Success;
    } else {
        return VersionConflict;
    }
}

ErrorStatus CObjectController::putValue(const std::string& obj_name, const std::string& br_name, const StoreObject& value,
                                        const std::string& commit_note, int revision_branch_previous,
                                        int& revision_branch_cur) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res ) {
        return res;
    }
    Branch branch;
    int revision_branch = NO_REVISION;
    res = getBranchInfo(object, br_name, branch, revision_branch);
    if ( Success != res ) {
        return res;
    }

    if ( revision_branch != revision_branch_previous ) {
        return RevisionConflict;
    }

    auto new_version = sha256(value.getSerializedValue());
    if ( new_version.compare(branch.cur_version) == 0 ) {
        return ValueNoChange;
    }
    Node node_pre;
    int revision_node_pre = NO_REVISION;
    res = _io_worker->loadNode(object.name, branch.cur_version, node_pre, revision_node_pre);
    if ( Success != res ) {
        return res;
    }
    res = _node_op->addChild(node_pre, new_version);
    if ( Success != res ) {
        return res;
    }
    Node node;
    res = _node_op->create(obj_name, new_version, commit_note, static_cast<int>(value.getType()), vector<string>{branch.cur_version}, node);
    if ( Success != res ) {
        return res;
    }
    res = _br_op->setVersion(branch, new_version);
    if ( Success != res ) {
        return res;
    }

    res = _io_worker->saveNode(node_pre, revision_node_pre);
    if ( Success != res ) {
        return res;
    }
    int revision_node_cur = NO_REVISION;
    res = _io_worker->createNode(node, revision_node_cur);
    if ( Success != res ) {
        return res;
    }
    res = _io_worker->saveBranch(branch, revision_branch_previous, revision_branch_cur);
    if ( Success != res ) {
        return res;
    }
    return _obj_io_worker->setValue(new_version, value);
}

ErrorStatus CObjectController::editObjectInfo(const std::string& obj_name, const std::string& note) {
    //    bool b_exist = false;
    //    auto res = this->isObjectExist(obj_name, b_exist);
    //    if ( Success != res )
    //        return res;
    //    if ( !b_exist )
    //        return DatasetNotExist;


    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    //    res = _ds_op->load(this->_io_worker, obj_name, object, revision_object);
    if ( Success != res )
        return res;
    res = _ds_op->setNote(object, note);
    if ( Success != res )
        return res;

    res = _io_worker->saveObject(object, revision_object);
    //    res = _ds_op->dump(_io_worker, object, revision_object);
    return res;
}



ErrorStatus CObjectController::editBranchInfo(const std::string& obj_name, const std::string& br_name,
                                              const std::string &note) {

    Branch branch;
    int revision = NO_REVISION;
    auto res = this->getBranchInfo(obj_name, br_name, branch, revision);
    if ( Success != res ) {
        return res;
    }
    res = _br_op->setNote(branch, note);
    if ( Success != res ) {
        return res;
    }
    int revision_new = NO_REVISION;
    res = _io_worker->saveBranch(branch, revision, revision_new);
    //    res = _br_op->dump(_io_worker, branch, revision);
    return res;
}

ErrorStatus CObjectController::resetBranchVersion(const std::string& obj_name, const std::string& br_name,
                                                  const std::string& version, int revision_pre, int& revision_new) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res )
        return res;
    Branch branch;
    int revision_branch = NO_REVISION;
    res = getBranchInfo(obj_name, br_name, branch, revision_branch);
    if ( Success != res )
        return res;

    if ( revision_branch != revision_pre ) {
        return RevisionConflict;
    }
    bool is_in = false;
    res = isVersionInBranch(object, branch, version, is_in);
    if ( Success != res )
        return res;
    if ( !is_in ) {
        return VersionNotInBranch;
    }

    res = _br_op->setVersion(branch, version);
    if ( Success != res ) {
        return res;
    }
    res = _io_worker->saveBranch(branch, revision_pre, revision_new);
    //    res =  _br_op->dump(_io_worker, branch, revision_pre);
    return res;
}

ErrorStatus CObjectController::getBranchVersions(const std::string& obj_name, const std::string& br_name,
                                                 std::vector<std::string>& versions, int& revision_branch) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res )
        return res;
    Branch branch;
    //    int revision_branch = NO_REVISION;
    res = getBranchInfo(obj_name, br_name, branch, revision_branch);
    if ( Success != res )
        return res;

    versions.clear();
    std::queue<std::string> queue_cycle;
    queue_cycle.push(branch.cur_version);
    while ( !queue_cycle.empty() ) {
        auto pn = queue_cycle.front();
        queue_cycle.pop();
        auto find_res = std::find(versions.begin(), versions.end(), pn);
        if ( find_res != versions.end() )
            continue;
        versions.emplace_back(pn);
        Node node;
        int revision_node = NO_REVISION;
        auto res_load_node = _io_worker->loadNode(obj_name, pn, node, revision_node);//_node_op->load(_io_worker, obj_name, pn, node, revision_node);
        if ( Success != res_load_node )
            continue;
        for ( auto &parent : node.parents ) {
            queue_cycle.push(parent);
        }
    }
    return Success;
}

ErrorStatus CObjectController::isVersionInBranch(const Object &object, const Branch& branch,
                                                 const std::string& version, bool &is_in) {
    is_in = false;
    std::queue<std::string> queue_cycle;
    queue_cycle.push(branch.cur_version);
    while ( !queue_cycle.empty() ) {
        auto pn = queue_cycle.front();
        queue_cycle.pop();
        if ( pn.compare(version) == 0 ) {
            is_in = true;
            break;
        }
        Node node;
        int revision_node = NO_REVISION;
        auto res_load_node = _io_worker->loadNode(object.name, pn, node, revision_node);//_node_op->load(_io_worker, object.name, pn, node, revision_node);
        if ( Success != res_load_node )
            continue;
        for ( auto &parent : node.parents ) {
            queue_cycle.push(parent);
        }
    }
    return Success;
}

ErrorStatus CObjectController::isVersionInBranch(const std::string& obj_name, const std::string& br_name,
                                                 const std::string& version, bool &is_in) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res )
        return res;
    Branch branch;
    int revision_branch = NO_REVISION;
    res = getBranchInfo(obj_name, br_name, branch, revision_branch);
    if ( Success != res )
        return res;

    return isVersionInBranch(object, branch, version, is_in);
}

ErrorStatus CObjectController::isVersionInObject(const Object &object, const std::string& version, bool &is_in) {
    Node node;
    int revision_node = NO_REVISION;
    auto res = getNodeInfo(object, version, node, revision_node);
    is_in = ( Success == res );
    //    is_in = ( object.nodes.end() != std::find(object.nodes.begin(), object.nodes.end(), version) );
    return Success;
}

ErrorStatus CObjectController::isVersionInObject(const std::string& obj_name, const std::string& version, bool &is_in) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res )
        return res;
    return isVersionInObject(object, version, is_in);
}


//ErrorStatus CObjectController::diffObjectVersion(const std::string& obj_name1, const std::string& version1,
//                                                 const std::string& obj_name2, const std::string& version2,
//                                                 const std::string& seperator, std::string& sres_diff) {
//    //    Dataset dataset1, dataset2;
//    Node node1, node2;
//    int revision_node1 = NO_REVISION;
//    int revision_node2 = NO_REVISION;
//    auto res = getNodeInfo(obj_name1, version1, node1, revision_node1);
//    if ( Success != res ) {
//        return res;
//    }
//    res = getNodeInfo(obj_name2, version2, node2, revision_node2);
//    if ( Success != res ) {
//        return res;
//    }
//    auto type1 = static_cast<StoreObjectType>(node1.type);
//    auto type2 = static_cast<StoreObjectType>(node2.type);
//    if ( type1 != type2 ) {
//        return DifferentTypeCannotDiff;
//    }
//    if ( type1 == StoreObjectType::Blob ) {
//        return TypeCannotDiff;
//    }

//    StoreObject value1(type1), value2(type1);
//    res = _obj_io_worker->getValue(version1, value1);
//    if ( Success != res ) {
//        return res;
//    }
//    res = _obj_io_worker->getValue(version2, value2);
//    if ( Success != res ) {
//        return res;
//    }
//    return CStoreObjectDiff::diff(value1, value2, sres_diff, seperator);
//}

//ErrorStatus CObjectController::diffObjectBranch(const std::string& obj_name1, const std::string& br_name1,
//                                                const std::string& obj_name2, const std::string& br_name2,
//                                                const std::string& seperator, std::string& sres_diff,
//                                                int& revision_branch1, int& revision_branch2) {
//    Branch branch1, branch2;
//    auto res = getBranchInfo(obj_name1, br_name1, branch1, revision_branch1);
//    if ( Success != res ) {
//        return res;
//    }
//    res = getBranchInfo(obj_name2, br_name2, branch2, revision_branch2);
//    if ( Success != res ) {
//        return res;
//    }
//    auto type1 = static_cast<StoreObjectType>(branch1.type);
//    auto type2 = static_cast<StoreObjectType>(branch2.type);
//    if ( type1 != type2 ) {
//        return DifferentTypeCannotDiff;
//    }
//    if ( type1 == StoreObjectType::Blob ) {
//        return TypeCannotDiff;
//    }

//    StoreObject value1(type1), value2(type1);
//    res = _obj_io_worker->getValue(branch1.cur_version, value1);
//    if ( Success != res ) {
//        return res;
//    }
//    res = _obj_io_worker->getValue(branch2.cur_version, value2);
//    if ( Success != res ) {
//        return res;
//    }
//    return CStoreObjectDiff::diff(value1, value2, sres_diff, seperator);
//}


ErrorStatus CObjectController::mergeBranch(const std::string& obj_name, const std::string& br_name,
                                           const std::string& refer_br_name, const std::string& commit_note,
                                           const std::string& seperator) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res ) {
        return res;
    }
    if ( StoreObjectType::String != static_cast<StoreObjectType>(object.type) ) {
        return BlobCannotMerge;
    }
    Branch branch_refer;
    int revision_branch = NO_REVISION;
    res = getBranchInfo(object, refer_br_name, branch_refer, revision_branch);
    if ( Success != res ) {
        return res;
    }
    return merge(object, br_name, branch_refer.cur_version, commit_note, seperator);
}

ErrorStatus CObjectController::mergeVersion(const std::string& obj_name, const std::string& br_name,
                                            const std::string& version, const std::string& commit_note,
                                            const std::string& seperator) {
    Object object;
    int revision_object = NO_REVISION;
    auto res = getObjectInfo(obj_name, object, revision_object);
    if ( Success != res ) {
        return res;
    }
    if ( StoreObjectType::String != static_cast<StoreObjectType>(object.type) ) {
        return BlobCannotMerge;
    }
    return merge(object, br_name, version, commit_note, seperator);
}

ErrorStatus CObjectController::merge(const Object& object, const std::string& br_name,
                                     const std::string& version_merge_in, const std::string& commit_note,
                                     const std::string& seperator) {
    Branch branch;
    int revision_branch = NO_REVISION;
    auto res = getBranchInfo(object, br_name, branch, revision_branch);
    if ( Success != res ) {
        return res;
    }
    auto branch_type = static_cast<StoreObjectType>(branch.type);
    if ( StoreObjectType::Blob == branch_type ) {
        return BlobCannotMerge;
    }
    Node node_br_cur, node_merge_in, node_ancestor;
    int revision_node_cur = NO_REVISION;
    res = getNodeInfo(object, branch.cur_version, node_br_cur, revision_node_cur);
    if ( Success != res ) {
        return res;
    }
    int revision_node_merge = NO_REVISION;
    res = getNodeInfo(object, version_merge_in, node_merge_in, revision_node_merge);
    if ( Success != res ) {
        return res;
    }
    auto merge_in_node_type = static_cast<StoreObjectType>(node_merge_in.type);
    if ( StoreObjectType::Blob == merge_in_node_type ) {
        return BlobCannotMerge;
    }
    res = findCommonAncestor(object, node_br_cur, node_merge_in, node_ancestor);
    if ( Success != res ) {
        return res;
    }
    string version_result;
    StoreObject value_cur(branch_type), value_merge(branch_type), value_ancestor(branch_type), value_result(branch_type);
    res = _obj_io_worker->getValue(branch.cur_version, value_cur);
    if ( Success != res ) {
        return res;
    }
    res = _obj_io_worker->getValue(version_merge_in, value_merge);
    if ( Success != res ) {
        return res;
    }
    res = _obj_io_worker->getValue(node_ancestor.version, value_ancestor);
    if ( Success != res ) {
        return res;
    }

    res = CStoreObjectMerge::merge(value_cur, value_merge, value_ancestor, value_result, seperator);
    if ( Success != res ) {
        return res;
    }
    version_result = sha256(value_result.getSerializedValue());
    res = _obj_io_worker->setValue(version_result, value_result);
    if ( Success != res ) {
        return res;
    }

    Node node_result;
    res = _node_op->create(object.name, version_result, commit_note, static_cast<int>(StoreObjectType::String), vector<string>{branch.cur_version, version_merge_in}, node_result);
    if ( Success != res ) {
        return res;
    }
    res = _br_op->setVersion(branch, version_result);
    if ( Success != res ) {
        return res;
    }

    int revision_node_result = NO_REVISION;
    res = _io_worker->createNode(node_result, revision_node_result);
    //    res = _node_op->dump(_io_worker, node_result, NO_REVISION, true);
    if ( Success != res ) {
        return res;
    }
    int revision_new = NO_REVISION;
    res = _io_worker->saveBranch(branch, revision_branch, revision_new);
    //    res = _br_op->dump(_io_worker, branch, revision_branch, false);
    if ( Success != res ) {
        return res;
    }

    return res;
}

ErrorStatus CObjectController::cycleChildren(const std::string& obj_name, const Node& node,
                                             map<string, dagNode*> &graph) {
    ErrorStatus res = Success;
    if ( graph[node.version] ) {
        return res;
    }
    graph[node.version] = new dagNode(node.version);
    for ( auto &child : node.children ) {
        if ( !graph[child] ) {
            dagNode *dchild = new dagNode(child);
            graph[child] = dchild;
        }
        //        graph[child]->parents.push_back(graph[node.version]);
        graph[node.version]->children.push_back(graph[child]);
    }
    for ( auto dchild : graph[node.version]->children ) {
        Node node_child;
        int revision_node = NO_REVISION;
        res = _io_worker->loadNode(obj_name, dchild->version, node_child, revision_node);//_node_op->load(_io_worker, obj_name, dchild->version, node_child, revision_node);
        if ( Success != res ) {
            return res;
        }
        res = cycleChildren(obj_name, node_child, graph);
        if ( Success != res ) {
            return res;
        }
    }
    return Success;
}

ErrorStatus CObjectController::findNode(const std::string& find_version, const std::string& cur_version,
                                        const std::string& prob_ancestor, map<string, dagNode*> &graph,
                                        bool &result, string& ancestor) {
    if ( result ) {
        return Success;
    }
    auto dnode = graph[cur_version];
    if ( !dnode ) {
        //        result = false;
        return Success;
    }
    if ( dnode->vis ) {
        //        result = false;
        return Success;
    }
    dnode->vis = true;
    if( dnode->version.compare(find_version) == 0 ) {
        ancestor = prob_ancestor;
        result = true;
    } else {
        for ( auto dchild : dnode->children ) {
            auto res = findNode(find_version, dchild->version, prob_ancestor, graph, result, ancestor);
            if ( res != Success )
                return res;
            if ( result ) {
                break;
            }
        }
    }
    return Success;
}

ErrorStatus CObjectController::findCommonAncestor(const Object &object, const Node &node1,
                                                  const Node& node2, Node& ancestor) {
    if ( node1.version.compare(node2.version) == 0 ) {
        ancestor = node1;
        return VersionSameError;
    }

    const Node *node_early, *node_later;
    if ( node1.date > node2.date ) {
        node_early = &node2;
        node_later = &node1;
    } else {
        node_early = &node1;
        node_later = &node2;
    }
    queue<string> queue_cycle_early;

    queue_cycle_early.push(node_early->version);
    //    for ( auto &parent_later : node_later->parents ) {
    //        queue_cycle_later.push(parent_later);
    //    }
    bool b_result = false;
    string s_ancestor;
    map<string, dagNode*> graph;
    while ( !queue_cycle_early.empty() ) {
        auto vn_early = queue_cycle_early.front();
        queue_cycle_early.pop();
        Node nn_early;
        int revision_node = NO_REVISION;
        auto res = _io_worker->loadNode(object.name, vn_early, nn_early, revision_node);//_node_op->load(_io_worker, object.name, vn_early, nn_early, revision_node);
        if ( Success != res ){
            return res;
        }
        res = cycleChildren(object.name, nn_early, graph);
        if ( Success != res ){
            return res;
        }
        res = findNode(node_later->version, vn_early, vn_early, graph, b_result, s_ancestor);
        if ( Success != res ){
            return res;
        }
        if ( b_result ) {
            break;
        }
        for ( auto &parent : nn_early.parents ) {
            queue_cycle_early.push(parent);
        }
    }
    for ( auto ite = graph.begin() ; ite!=graph.end() ; ++ite ) {
        delete ite->second;
        ite->second = nullptr;
    }
    int revision_node = NO_REVISION;
    auto res = _io_worker->loadNode(object.name, s_ancestor, ancestor, revision_node);//_node_op->load(_io_worker, object.name, s_ancestor, ancestor, revision_node);
    if ( Success != res ){
        return res;
    }
    return Success;
}


}
